/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.seatunnel.api.source;

import org.apache.seatunnel.api.common.PluginIdentifierInterface;
import org.apache.seatunnel.api.common.SeaTunnelPluginLifeCycle;
import org.apache.seatunnel.api.serialization.DefaultSerializer;
import org.apache.seatunnel.api.serialization.Serializer;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;

import java.io.Serializable;
import java.util.List;

/**
 * The interface for Source. It acts like a factory class that helps construct the {@link
 * SourceSplitEnumerator} and {@link SourceReader} and corresponding serializers.
 *
 * @param <T> The type of records produced by the source.
 * @param <SplitT> The type of splits handled by the source.
 * @param <StateT> The type of checkpoint states.
 */
public interface SeaTunnelSource<T, SplitT extends SourceSplit, StateT extends Serializable>
        extends Serializable,
                PluginIdentifierInterface,
                SeaTunnelPluginLifeCycle,
                SeaTunnelJobAware {

    /**
     * Get the boundedness of this source.
     *
     * @return the boundedness of this source.
     */
    Boundedness getBoundedness();

    /**
     * Get the data type of the records produced by this source.
     *
     * @deprecated Please use {@link #getProducedCatalogTables}
     * @return SeaTunnel data type.
     */
    @Deprecated
    default SeaTunnelDataType<T> getProducedType() {
        return (SeaTunnelDataType) getProducedCatalogTables().get(0).getSeaTunnelRowType();
    }

    /**
     * Get the catalog tables output by this source, It is recommended that all connectors implement
     * this method instead of {@link #getProducedType}. CatalogTable contains more information to
     * help downstream support more accurate and complete synchronization capabilities.
     */
    default List<CatalogTable> getProducedCatalogTables() {
        throw new UnsupportedOperationException(
                "getProducedCatalogTables method has not been implemented.");
    }

    /**
     * Create source reader, used to produce data.
     *
     * @param readerContext reader context.
     * @return source reader.
     * @throws Exception when create reader failed.
     */
    SourceReader<T, SplitT> createReader(SourceReader.Context readerContext) throws Exception;

    /**
     * Create split serializer, use to serialize/deserialize split generated by {@link
     * SourceSplitEnumerator}.
     *
     * @return split serializer.
     */
    default Serializer<SplitT> getSplitSerializer() {
        return new DefaultSerializer<>();
    }

    /**
     * Create source split enumerator, used to generate splits. This method will be called only once
     * when start a source.
     *
     * @param enumeratorContext enumerator context.
     * @return source split enumerator.
     * @throws Exception when create enumerator failed.
     */
    SourceSplitEnumerator<SplitT, StateT> createEnumerator(
            SourceSplitEnumerator.Context<SplitT> enumeratorContext) throws Exception;

    /**
     * Create source split enumerator, used to generate splits. This method will be called when
     * restore from checkpoint.
     *
     * @param enumeratorContext enumerator context.
     * @param checkpointState checkpoint state.
     * @return source split enumerator.
     * @throws Exception when create enumerator failed.
     */
    SourceSplitEnumerator<SplitT, StateT> restoreEnumerator(
            SourceSplitEnumerator.Context<SplitT> enumeratorContext, StateT checkpointState)
            throws Exception;

    /**
     * Create enumerator state serializer, used to serialize/deserialize checkpoint state.
     *
     * @return enumerator state serializer.
     */
    default Serializer<StateT> getEnumeratorStateSerializer() {
        return new DefaultSerializer<>();
    }
}
